package link.jfire.jnet.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import link.jfire.jnet.common.InternalResultContainer;
import link.jfire.jnet.common.buffer.ByteBuf;
import link.jfire.jnet.common.buffer.DirectByteBuf;
import link.jfire.jnet.common.buffer.DirectByteBufPool;
import link.jfire.jnet.common.decodec.FrameDecodec;
import link.jfire.jnet.common.exception.SocketException;
import link.jfire.jnet.common.handler.DataHandler;

/**
 * 客户端工具类，注意，该客户端是非线程安全类，其方法不可多线程运行
 * 
 * @author linbin
 *         
 */
public abstract class AioClient
{
    protected long                      readTimeout  = 3000;
    // 默认的超时等待时间是30分钟
    protected long                      waitTimeout  = 1000 * 60 * 30;
    protected String                    address;
    protected int                       port;
    protected AsynchronousSocketChannel socketChannel;
    protected AsynchronousChannelGroup  channelGroup;
    protected FrameDecodec              frameDecodec;
    protected DataHandler[]             readHandlers;
    protected DataHandler[]             writeHandlers;
    protected DirectByteBuf             ioBuf        = DirectByteBufPool.getInstance().get(100);
    protected ClientReadCompleter       readCompleter;
    protected volatile boolean          connectState = UNCONNECTED;
    public static final boolean         UNCONNECTED  = false;
    public static final boolean         CONNECTED    = true;
    protected Future<Void>              NORESULT     = new Future<Void>() {
                                                         
                                                         @Override
                                                         public boolean cancel(boolean mayInterruptIfRunning)
                                                         {
                                                             return false;
                                                         }
                                                         
                                                         @Override
                                                         public boolean isCancelled()
                                                         {
                                                             return false;
                                                         }
                                                         
                                                         @Override
                                                         public boolean isDone()
                                                         {
                                                             return true;
                                                         }
                                                         
                                                         @Override
                                                         public Void get() throws InterruptedException, ExecutionException
                                                         {
                                                             return null;
                                                         }
                                                         
                                                         @Override
                                                         public Void get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
                                                         {
                                                             return null;
                                                         }
                                                     };
                                                     
    public AioClient setReadTimeout(long readTimeout)
    {
        this.readTimeout = readTimeout;
        return this;
    }
    
    public AioClient setWaitTimeout(long waitTimeout)
    {
        this.waitTimeout = waitTimeout;
        return this;
    }
    
    public AioClient setAddress(String address)
    {
        this.address = address;
        return this;
    }
    
    public AioClient setPort(int port)
    {
        this.port = port;
        return this;
    }
    
    public AioClient setChannelGroup(AsynchronousChannelGroup channelGroup)
    {
        this.channelGroup = channelGroup;
        return this;
    }
    
    public AioClient setReadHandlers(DataHandler... readHandlers)
    {
        this.readHandlers = readHandlers;
        return this;
    }
    
    public AioClient setWriteHandlers(DataHandler... writeHandlers)
    {
        this.writeHandlers = writeHandlers;
        return this;
    }
    
    public AioClient setFrameDecodec(FrameDecodec frameDecodec)
    {
        this.frameDecodec = frameDecodec;
        return this;
    }
    
    public void connect() throws IOException, InterruptedException, ExecutionException
    {
        socketChannel = AsynchronousSocketChannel.open(channelGroup);
        socketChannel.connect(new InetSocketAddress(address, port)).get();
        readCompleter = new ClientReadCompleter(frameDecodec, readHandlers);
        readAndWait();
        connectState = CONNECTED;
    }
    
    protected void continueRead()
    {
        socketChannel.read(getReadBuffer(), readTimeout, TimeUnit.MILLISECONDS, this, readCompleter);
    }
    
    protected void readAndWait()
    {
        socketChannel.read(getReadBuffer(), waitTimeout, TimeUnit.MILLISECONDS, this, readCompleter);
        
    }
    
    private ByteBuffer getReadBuffer()
    {
        ioBuf.compact();
        ByteBuffer ioBuffer = ioBuf.nioBuffer();
        ioBuffer.position(ioBuffer.limit()).limit(ioBuffer.capacity());
        return ioBuffer;
    }
    
    /**
     * 将一个对象写出并且返回一个future。该future表明的是服务端对该请求报文的响应报文的处理结果
     * 
     * @param object
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws SocketException
     */
    public Future<?> write(Object object) throws InterruptedException, ExecutionException, SocketException
    {
        return write(object, 0);
    }
    
    /**
     * 将一个对象写出并且指定开始处理时的handler顺序，然后返回一个future。该future表明的是服务端对该请求报文的响应报文的处理结果
     * 
     * @param object
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws SocketException
     */
    public Future<?> write(Object data, int index) throws InterruptedException, ExecutionException, SocketException
    {
        if (connectState == UNCONNECTED)
        {
            throw new InterruptedException("链接已经中断，请重新链接后再发送信息");
        }
        ClientInternalResult internalResult = InternalResultContainer.buildClient(data, this, index);
        for (int i = index; i < writeHandlers.length;)
        {
            data = writeHandlers[i].handle(data, internalResult);
            if (i == internalResult.getIndex())
            {
                i++;
                internalResult.setIndex(i);
            }
            else
            {
                i = internalResult.getIndex();
            }
            
        }
        if (data instanceof ByteBuf<?>)
        {
            Future<?> result = buildFuture();
            ByteBuf<?> buf = (ByteBuf<?>) data;
            int writed = 0;
            while (buf.remainRead() > 0)
            {
                writed = socketChannel.write(((ByteBuf<?>) data).nioBuffer()).get();
                buf.addReadIndex(writed);
            }
            buf.release();
            return result;
        }
        else
        {
            return NORESULT;
        }
    }
    
    protected abstract Future<?> buildFuture();
    
    protected abstract void popOneFuture(Object obj);
    
    protected abstract void releaseAllFuture(Throwable e);
    
    public void close(Throwable e)
    {
        try
        {
            connectState = UNCONNECTED;
            socketChannel.close();
            ioBuf.release();
        }
        catch (IOException e1)
        {
            e.printStackTrace();
        }
        releaseAllFuture(e);
    }
}
